package com.pxene.dmp.mapper;

import java.io.IOException;
import java.net.MalformedURLException;
import java.net.URL;
import java.net.URLDecoder;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import com.pxene.dmp.common.URLUtils;

public class GetSearchWordMapper extends
		Mapper<LongWritable, Text, Text, Text> {

	private URL urlParser;
	private Text resultKey = new Text();
	private Text resultVal = new Text();
	
	@Override
	protected void setup(Context context) throws IOException, InterruptedException {
		Configuration conf = context.getConfiguration();
		// 设置配置文件名
		System.setProperty("config-file", conf.get("config-file"));
	}
	
	@Override
	protected void map(LongWritable key, Text value, Context context)
			throws IOException, InterruptedException {
		String[] columns = value.toString().split("\\|");
		if (columns.length != 7) {
			return;
		}
		String imsi = columns[0];
		String bsid = columns[1];
		String starttime = columns[2];
		String url = columns[3];
		String province = columns[4];
		String dataType = columns[5];
		
		// url解码
		url = URLDecoder.decode(url.replaceAll("%(?![0-9a-fA-F]{2})", "%25"), "GB2312");
		try {
			urlParser = new URL(url);
		} catch(MalformedURLException e) {
			return;
		}
		
		String word = "";
		Map<String, String> map = URLUtils.getQueryMap(urlParser.getQuery());
		if (map.containsKey("q")) {
			word = map.get("q").trim();
		}
		if (map.containsKey("wd")) {
			word = map.get("wd").trim();
		}
		if (map.containsKey("kw")) {
			word = map.get("kw").trim();
		}
		if (map.containsKey("word")) {
			word = map.get("word").trim();
		}
		if (map.containsKey("query")) {
			word = map.get("query").trim();
		}
		if (map.containsKey("keyword")) {
			word = map.get("keyword").trim();
		}
		
		if (word.length() >= 3 && word.length() <= 40) {
			resultKey.set(imsi + "|" + bsid + "|" + starttime + "|" + province + "|" + dataType);
			resultVal.set(URLUtils.removePunctuations(word));
			context.write(resultKey, resultVal);
		}
		
	}
	
}
